package com.sky.business.sys.queue;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sky.business.sys.domain.MQTTMessage;
import com.sky.business.sys.service.MQTTMessageService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.SimpleTimeZone;

/**
 * mqtt消息处理
 */
@Slf4j
public class MessageConsumerRunnable implements Runnable{

    private final MQTTMessageService mqttMessageService;

    public static ObjectMapper objectMapper = new ObjectMapper().setTimeZone(SimpleTimeZone.getTimeZone("GMT+8"));


    public MessageConsumerRunnable(MQTTMessageService mqttMessageService) {
        this.mqttMessageService = mqttMessageService;
    }

    @SneakyThrows
    @Override
    public void run() {
        int heartbeatCount =0;
        int nodeReportCount = 0;
        while (true){
            try {
                MQTTMessage mqttMessage = MQTTMessageQueueConfig.poll();
                if (mqttMessage != null) {
                    log.debug("从队列中消费了MQTTMessage,消息类型：{}, 消息时间：{}", mqttMessage.getType().name(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(LocalDateTime.now()));
                    String msg = mqttMessage.getMessage();
                    try {
                        JsonNode jsonNode = objectMapper.readTree(msg);
                        if (jsonNode.has("event")) {
                            String event = jsonNode.get("event").asText();
                            if("HeartBeat".equals(event)){
                                if(++heartbeatCount < 10){
                                    continue;
                                }else{
                                    heartbeatCount = 0;
                                }
                            }
                            if("nodeReport".equals(event)){
                                if(++nodeReportCount <12){
                                    continue;
                                }else{
                                    nodeReportCount = 0;
                                }
                            }
                            mqttMessage.setEvent(event);
                        }
                        if (jsonNode.has("metadata")) {
                            JsonNode metadata = jsonNode.get("metadata");
                            if (metadata.has("taskId") && !metadata.get("taskId").isNull()) {
                                mqttMessage.setTaskId(metadata.get("taskId").asText());
                            }
                            if (metadata.has("deviceId") && !metadata.get("deviceId").isNull()) {
                                mqttMessage.setDeviceId(metadata.get("deviceId").asText());
                            }
                        }

                        if (jsonNode.has("nodeId")) {
                            try {
                                mqttMessage.setNodeId(jsonNode.get("nodeId").asInt());
                            } catch (Exception e1) {
                                log.error("nodeId解析发生异常", e1);
                            }
                        }
                        if (jsonNode.has("timestamp")) {
                            mqttMessage.setCreateDate(new Date(jsonNode.get("timestamp").asLong()));
                        }
                    } catch (Exception e) {
                        log.error("解析mqtt消息内容发生异常,msg:{}", msg, e);
                    }
                    mqttMessageService.save(mqttMessage);
                }else {
                    Thread.sleep(1000);
                }
            }catch (Exception e){
                log.error("mqtt消息保存发生异常",e);
            }
        }
    }
}
